package commands

import (
	"encoding/json"
	"fmt"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"github.com/sirupsen/logrus"
	"github.com/tluo-github/super-runner/common"
	"github.com/tluo-github/super-runner/helpers/kafka"
	prometheus_helper "github.com/tluo-github/super-runner/helpers/prometheus"
	"github.com/urfave/cli"
	"net"
	"net/http"
	"os"
	"os/signal"
	"syscall"
)

type RunCommand struct {
	config common.RunnerConfig 		// super runner 配置文件
	jobCollector *prometheus_helper.JobCollector  	//  prometheus  job 信息收集器
	versionCollector *prometheus_helper.VersionCollector // prometheus version 收集器
	kafkaHelper *kafka.KafkaHelper // kafka helper 类

	//stopSignals将捕获通知处理的信号：SIGTERM，SIGQUIT，中断，Kill
	stopSignals chan os.Signal
	stopSignal os.Signal

}

func (mr *RunCommand) log() *logrus.Entry {
	return logrus.WithField("runner-name", mr.config.Name)
}



func (mr *RunCommand) Execute(cmdcontext *cli.Context) {

	logrus.Info("Run Super Runner.")
	// 加载配置文件
	mr.loadConfig(cmdcontext.String("config"))
	// 设置 metrics
	mr.setupMetricsAndDebugServer()
	mr.stopSignals = make(chan os.Signal)
	go mr.handleGracefulShutdown()
	// 运行 super-runner
	mr.run()


}

// 业务逻辑入口函数
func (mr *RunCommand) run() {


	mr.kafkaHelper = kafka.GetKafkaHelper(mr.config.KafkaConfig)
	defer mr.kafkaHelper.Close()

	// 并发限制队列
	queue := make(chan int,mr.config.Concurrent)
	for mr.stopSignal == nil{
		// 入队列
		queue <- 1
		var build common.Build
		err := json.Unmarshal(mr.kafkaHelper.ReadMessage(),&build)
		if err != nil {
			logrus.Warn(fmt.Sprintf("json unmarshall faild: %v", err))
			// 出队列
			<-queue
		} else {
			go func() {
				new_job := common.Job{
					Runner: mr.config,
					Build:  build,
				}
				// promethous metrics
				mr.jobCollector.AddBuild(&build)
				defer mr.jobCollector.RemoveBuild(&build)
				err := new_job.Run()
				if err != nil{
					// promethous metrics
					mr.jobCollector.FailureBuild()
				}

				// 出队列
				<-queue
			}()
		}
	}


}

// 优雅停机
func (mr *RunCommand) handleGracefulShutdown() error {
	signal.Notify(mr.stopSignals, syscall.SIGQUIT, syscall.SIGTERM, os.Interrupt, os.Kill)
	mr.stopSignal = <-mr.stopSignals
	mr.log().
		WithField("StopSignal", mr.stopSignal).
		Warning("Starting graceful shutdown, waiting for builds to finish")
	os.Exit(0)
	return nil
}
// loadConfig 方法加载解析super runner 配置文件
func (mr *RunCommand) loadConfig(config_file string) {
	err := mr.config.LoadConfig(config_file)
	if err != nil {
		mr.log().Fatal(err)
	}
}

func (mr *RunCommand) setupMetricsAndDebugServer() {
	listenAddress, err := mr.config.GetListenAddress()
	if err != nil {
		mr.log().Errorf("invalid listen address: %s", err.Error())
		return
	}

	listener, err := net.Listen("tcp", listenAddress)
	if err != nil {
		mr.log().WithError(err).Fatal("Failed to create listener for metrics server")
	}

	mux := http.NewServeMux()

	go func() {
		err := http.Serve(listener, mux)
		if err != nil {
			mr.log().WithError(err).Fatal("Metrics server terminated")
		}
	}()

	mr.serveMetrics(mux)
	mr.log().
		WithField("address", listenAddress).
		Info("Metrics server listening")
}

func (mr *RunCommand) serveMetrics(mux *http.ServeMux) {
	registry := prometheus.NewRegistry()
	// 暴露 程序版本
	registry.MustRegister(mr.versionCollector)
	// 暴露 Super Runner job 相关指标信息
	registry.MustRegister(mr.jobCollector)

	// 暴露关于进程的Go-specific指标(GC stats, goroutines，等等)。
	//registry.MustRegister(prometheus.NewGoCollector())
	// 暴露与go无关的进程指标(内存使用、文件描述符等)。
	//registry.MustRegister(prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))

	mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))

}

func init() {
	common.RegisterCommand("run","run super runner",&RunCommand{
		jobCollector: prometheus_helper.NewJobCollector(),
		versionCollector: prometheus_helper.NewVersionCollector(),
	} )

}